package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.SAN061;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.URLUtil;

import scala.Tuple2;

/**
 * 网站后缀访问量
 * 本省接入网站各域名后缀访问量变化统计（要求是省内接入网站，即dip在BR2002中存在）
 */
public class DomainAnalyzTask_dayNew  implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_dayNew.class);
	
	private static final long serialVersionUID = 1L;
	private String outName;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		log.info("==============DomainAnalyzTask begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame dipFrame = dataFrames.get("dws_hour_san061");
			String sql="select buss_value,sum(count_value) as count from dws_hour_san061_temp where buss_type='WEBSITE_COUNT_SUFFIX' and sta_range=0 group by  buss_value";
			DataFrame dataSource=dipFrame.sqlContext().sql(sql);
			JavaRDD<Map<String, String>> mapRDD = dataSource.toJavaRDD().map(new Function<Row,Map<String,String>>(){

				@Override
				public Map<String, String> call(Row row) throws Exception {
					Map<String, String> map=new HashMap<String,String>();
					String buss_value=row.getAs("buss_value").toString();
					 String count=row.getAs("count").toString();
					map.put("buss_value", buss_value);
					map.put("count",count);
					
					 return  map;
				}
				
			}); 
			
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(),dayStr);
	        JavaRDD<SAN061> rdd = createRdd(mapRDD, date);
	    	DataFrame df = dipFrame.sqlContext().createDataFrame(rdd, SAN061.class);
	    	//df.registerTempTable(outName); //插入到hive中需要使用临时表
	    	outList.add(new OutQueueEntity(outName, df));
	    	
	    	CommonUtils.deleteTaskTableData("SAN061", date, "WEBSITE_COUNT_SUFFIX", 0, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("DomainAnalyzTask_New1111网站后缀访问量:exectime: "+min+" min............");
		
		return outList;
	}
	
	@SuppressWarnings("serial")
	private JavaRDD<SAN061> createRdd(JavaRDD<Map<String, String>> mapRDD, final Date date) {
		JavaRDD<SAN061> rdd = mapRDD.mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN061>() {

			@Override
			public Iterable<SAN061> call(Iterator<Map<String,String>> it){
				List<SAN061> list = new ArrayList<SAN061>();
				while(it.hasNext()){
					Map<String,String> map = it.next();
					String domain = map.get("buss_value").toString();
					long count=Long.parseLong(map.get("count").toString());
					if(StringUtils.isNotBlank(domain)){
		        		SAN061 sa = new SAN061();
		        		sa.setYear(DateUtil.getCurrentYear(date));
		        		sa.setHalf_year(DateUtil.getHalfYear(date));
		        		sa.setQuarter(DateUtil.getQuarter(date));
		        		sa.setMonth(DateUtil.getCurrentMonth(date));
		        		sa.setWeek(DateUtil.getCurrentWeek(date));
		        		sa.setDay(DateUtil.getCurrentDay(date));
		        		sa.setBuss_type("WEBSITE_COUNT_SUFFIX");
		        		sa.setBuss_value(domain);
		        		sa.setCount_value(count);
		        		sa.setSta_range(0);
		        		list.add(sa);
		        	}
				}
				return list;
			}
		});
		return rdd;
	}

	
	}


